#include "view.h"
#include "view_protocol.pb.h"
#include "method_thread.h"
#include "timeutils.h"
#include "trace.h"

#include <sstream>
#include <algorithm>
#include <assert.h>
#include <sys/time.h>
#include <time.h>
#include <pthread.h>

using namespace std;

/**
 * Create a new view manager.
 * @param primary the id of the primary node
 * @param m the underlying messaging object
 * @param obs pointer to the view observer object
 */
viewmgr::viewmgr(const string &primary, messaging *m, view_observer *obs)
    : primary_(primary), me_(m->id()), vid_(0), stable_(false), hermes_(m),
      pxs_(primary, m, this), vobserver_(obs)
{
    pthread_mutex_init(&last_seen_m_, NULL);
    pthread_mutex_init(&view_m_, NULL);
    pthread_cond_init(&joined_cv_, NULL);
    pthread_cond_init(&stable_cv_, NULL);
    pthread_cond_init(&recovery_cv_, NULL);
    pthread_cond_init(&heartbeat_cv_, NULL);
    pthread_cond_init(&viewchange_cv_, NULL);

    hermes_->subscribe("VIEW.HEARTBEAT", this);
    hermes_->subscribe("VIEW.OUTDATED", this);
    hermes_->subscribe("VIEW.JOINREQ", this);
    hermes_->subscribe("VIEW.JOINRES", this);
    hermes_->subscribe("VIEW.REMOVEREQ", this);

    // TODO read the log first to recover from a (possible) previous failure
    // this would set vid_ properly

    if (vid_ == 0) {
        vector<string> mem;
        if (me_ == primary) {
            // I am started as the coordinator, so I can move onto the 2st view
            TINFO("%s started as the primary", me_.c_str());
            vector<string> empty;
            empty.clear();
            switch_view(1, me_, empty);
            viewchange_th_ = method_thread(this, false, &viewmgr::viewchange);
            assert (viewchange_th_ != 0);
        } else {
            recovery_th_ = method_thread(this, false, &viewmgr::recovery);
            assert (recovery_th_ != 0);
        }
    }

    heartbeat_th_ = method_thread(this, false, &viewmgr::heartbeater);
    assert (heartbeat_th_ != 0);
}

viewmgr::~viewmgr()
{
    pthread_mutex_destroy(&last_seen_m_);
    pthread_mutex_destroy(&view_m_);
    pthread_cond_destroy(&joined_cv_);
    pthread_cond_destroy(&stable_cv_);
    pthread_cond_destroy(&heartbeat_cv_);
    pthread_cond_destroy(&viewchange_cv_);
    pthread_cond_destroy(&recovery_cv_);
}

/**
 * Return the members in the replication group. The first element is always
 * the primary.
 * Assumes view_m_ is held.
 */
vector<string> viewmgr::members()
{
    vector<string> m = cohorts_;
    m.insert(m.begin(), primary_);
    return m;
}

/**
 * Remove a node's membership.
 * @param m the node to be removed
 */
void viewmgr::remove(const string &m)
{
    pthread_mutex_lock(&view_m_);
    if (m == primary_) {
        // oops! the primary is down. in the current setup, all we can do is
        // to mark the view as unstable and wait for the primary to come back
    } else {
        if (me_ == primary_) {
            // the primary can propose a new view to eliminate the dead node
        } else {
            // non-primary nodes delegate to the primary
        }
    }
    pthread_mutex_unlock(&view_m_);
}

/**
 * Obtain the current view.
 */
void viewmgr::get_view(unsigned *vid, string *primary,
        vector<string> *cohorts) const
{
    pthread_mutex_lock(&view_m_);
    if (vid) {
        *vid = vid_;
    }
    if (primary) {
        *primary = primary_;
    }
    if (cohorts) {
        cohorts->assign(cohorts_.begin(), cohorts_.end());
    }
    pthread_mutex_unlock(&view_m_);
}

/**
 * Callback on paxos commit. View manager uses this to update its current view.
 */
void viewmgr::paxos_commit(unsigned instance, const string &chosen)
{
    pthread_mutex_lock(&view_m_);
    // we are being notified that a new view has been decided by paxos
    unsigned vid;
    string primary;
    vector<string> cohorts;
    if (parse_str(chosen, &vid, &primary, &cohorts)) {
        if (me_ != primary && !isamember(me_, cohorts)) {
            // i am not in the new view; start the joining process
            pthread_cond_signal(&recovery_cv_);
        }
        switch_view(vid, primary, cohorts);
    }
    pthread_mutex_unlock(&view_m_);
}

/**
 * Repeatedly send heartbeats to appropriate audience.
 */
void viewmgr::heartbeater()
{
    struct timeval now;
    struct timespec until;

    pthread_mutex_lock(&view_m_);
    while (true) {
        check_liveness();

        while (!stable_) {
            pthread_cond_wait(&stable_cv_, &view_m_);
        }

        gettimeofday(&now, NULL);
        until.tv_sec = now.tv_sec + 5;
        until.tv_nsec = 0;
        pthread_cond_timedwait(&heartbeat_cv_, &view_m_, &until);

        if (!isamember(me_, cohorts_)) {
            // oops, i'm not enlisted in the group; skip this round
            TINFO("i (%s) am not a member yet; starting recovery", me_.c_str());
            pthread_cond_signal(&recovery_cv_);
            continue;
        }

        // if i am the primary, i should send heartbeats to all the other
        // cohorts; if i am not, i just send a heartbeat to the primary
        DataMessage wrapper;
        Heartbeat heartbeat;
        wrapper.set_subject("VIEW.HEARTBEAT");
        heartbeat.set_vid(vid_);
        wrapper.set_body(heartbeat.SerializeAsString());
        if (me_ == primary_) {
            for (unsigned i = 0; i < cohorts_.size(); ++i) {
                wrapper.clear_dest();
                wrapper.add_dest(cohorts_[i]);
                TDEBUG("primary sending heartbeat to %s",cohorts_[i].c_str());
                hermes_->deliver(&wrapper);
            }
        } else {
            TDEBUG("sending heartbeat to primary %s", primary_.c_str());
            wrapper.add_dest(primary_);
            hermes_->deliver(&wrapper);
        }
    }
    pthread_mutex_unlock(&view_m_);
}

/**
 * Check if the current node is a member of the network. And if not, try to
 * join via the primary. In the future, this function should also synchronize
 * its state with the primary for it may be one or more steps behind other
 * peers after it joins the network.
 */
void viewmgr::recovery()
{
    pthread_mutex_lock(&view_m_);
    while (true) {
        while (!isamember(me_, cohorts_)) {
            TDEBUG("starting recovery. join via primary %s", primary_.c_str());
            if (join()) {
                TINFO("%s joined via %s", me_.c_str(), primary_.c_str());
            } else {
                // sleep for a while and then retry
                pthread_mutex_unlock(&view_m_);
                sleep(10);
                pthread_mutex_lock(&view_m_);
            }
        }
        
        TDEBUG("recovery thread goes to sleep");
        pthread_cond_wait(&recovery_cv_, &view_m_);
    }
    pthread_mutex_unlock(&view_m_);
}

/**
 * Handle the view change queue. This function is only called on the primary.
 * Currently, only one kind of view change is
 * supported, that is newly joining node. We should also support removal of a
 * node.
 */
void viewmgr::viewchange()
{
    pthread_mutex_lock(&view_m_);
    while (true) {
        while (vc_q_.empty()) {
            TDEBUG("waiting for new joining node");
            pthread_cond_wait(&viewchange_cv_, &view_m_);
        }
        const pair<string, bool> item = vc_q_.front();
        const string &peer = item.first;
        TDEBUG("retrieving front of join queue: %s", peer.c_str());
        vc_q_.pop_front();
        if (item.second) {
            do_join(peer);
        } else {
            do_remove(peer);
        }
    }

    pthread_mutex_unlock(&view_m_);
}

// assuming view_m_ is held
void viewmgr::do_join(const string &peer)
{
    if (isamember(peer, cohorts_)) {
        TINFO("%s has already joined", peer.c_str());
        // return my current view anyway. this is probably a duplicate
        // join request because of retransmission
        DataMessage resp;
        JoinResponse jresp;
        resp.set_subject("VIEW.JOINRES");
        resp.add_dest(peer);
        jresp.set_joined(true);
        jresp.set_vid(vid_);
        jresp.set_primary(primary_);
        for (unsigned i = 0; i < cohorts_.size(); ++i) {
            jresp.add_cohorts(cohorts_[i]);
        }
        resp.set_body(jresp.SerializeAsString());
        hermes_->deliver(&resp);
    } else {
        if (vobserver_) {
            vobserver_->previewchange(vid_);
        }
        vector<string> new_mems = members();
        TDEBUG("cur view:\n>>>\n%s\n<<<", valueof(vid_, new_mems).c_str());
        new_mems.push_back(peer);
        if (new_mems.size() > 2) {
            sort(new_mems.begin() + 1, new_mems.end());
        }
        // since i am the distinguished proposer, it is impossible that
        // other values can be chosen
        TINFO("starting paxos for new view:\n>>>\n%s\n<<<",
                valueof(vid_ + 1, new_mems).c_str());
        // temporarily dropping view mutex
        pthread_mutex_unlock(&view_m_);
        string v = pxs_.start(members(), valueof(vid_ + 1, new_mems));

        pthread_mutex_lock(&view_m_);
        TINFO("new view: %s at %u", v.c_str(), vid_);

        // we need to tell the joining node about our current view, because
        // it does not get the paxos decide message before it "officially"
        // becomes a member of the family
        DataMessage resp;
        JoinResponse jresp;
        resp.set_subject("VIEW.JOINRES");
        resp.add_dest(peer);
        jresp.set_joined(isamember(peer, cohorts_));
        jresp.set_vid(vid_);
        jresp.set_primary(primary_);
        for (unsigned i = 0; i < cohorts_.size(); ++i) {
            jresp.add_cohorts(cohorts_[i]);
        }
        resp.set_body(jresp.SerializeAsString());
        TINFO("sending %s join reply", peer.c_str());
        hermes_->deliver(&resp);
    }
}

// assuming view_m_ is held
void viewmgr::do_remove(const string &peer)
{
    assert(me_ == primary_); // only called on the primary
 
    if (peer == me_) {
        // this is really weird... shouldn't happen anyway
        return;
    }

    if (!isamember(peer, cohorts_)) {
        TWARN("%s is not in view; couldn't remove it", peer.c_str());
        return;
    }

    if (vobserver_) {
        vobserver_->previewchange(vid_);
    }
    vector<string> new_mems = members();
    TDEBUG("cur view:\n>>>\n%s\n<<<", valueof(vid_, new_mems).c_str());
    vector<string>::iterator peer_pos;
    if ((peer_pos = find(cohorts_.begin(), cohorts_.end(),
                    peer)) != cohorts_.end()) {
        new_mems.erase(peer_pos);
    }
    string new_mems_str = valueof(vid_ + 1, new_mems);
    TINFO("starting paxos for new view:\n>>>\n%s\n<<<", new_mems_str.c_str());
    // temporarily dropping view mutex
    pthread_mutex_unlock(&view_m_);
    string v = pxs_.start(members(), new_mems_str);

    // we have to reobtain the lock for the caller of this function
    pthread_mutex_lock(&view_m_);
    TINFO("new view: %s at %u", v.c_str(), vid_);

    // we don't have to anything here because paxos will sort that out
}

// assumes view_m_ is held
bool viewmgr::join()
{
    struct timespec now, next_deadline, firm_deadline;
    int delta = 5000; // initial timeout was 5 seconds

    DataMessage msg;
    msg.set_subject("VIEW.JOINREQ");
    msg.add_dest(primary_);
    ostringstream oss;
    oss << vid_;
    msg.set_body(oss.str());

    clock_gettime(CLOCK_REALTIME, &now);
    add_timespec(now, 120000, &firm_deadline); // a maximum of 2-min timeout

    while (true) {
        if (!firm_deadline.tv_sec) {
            break;
        }
        clock_gettime(CLOCK_REALTIME, &now);
        add_timespec(now, delta, &next_deadline);
        if (cmp_timespec(next_deadline, firm_deadline) > 0) {
            next_deadline = firm_deadline;
            // mark for the next iteration to indicate that we can abort
            firm_deadline.tv_sec = 0;
        }
        TDEBUG("sending join request to %s", primary_.c_str());
        hermes_->deliver(&msg);
        if (pthread_cond_timedwait(&joined_cv_, &view_m_, &next_deadline) ==
                ETIMEDOUT) {
            TWARN("join times out after %d ms", delta);
        } else {
            TDEBUG("successfully joined. join() returns now");
            break;
        }
        delta <<= 1;
    }

    // TODO if successfully joined, replay the log returned by the primary
    return isamember(me_, cohorts_);
}

/**
 * Check if a specified node is a member of a group.
 * assumes view_m_ is held.
 * @param m the id of the node to test
 * @param sorted a sorted vector containing peer IDs
 */
bool viewmgr::isamember(const string &m, const vector<string> &sorted)
{
    return m == primary_ || binary_search(sorted.begin(), sorted.end(), m);
}

/**
 * Check if the current view is stable.
 * If a previous operation has already set the stable flag to false, this
 * function simply returns false. Otherwise, it checks if any peer that is
 * supposed to be in the current view has ceased to send heartbeats for a
 * certain period of time.
 * This function assumes the view_m_ lock is held.
 */
void viewmgr::check_liveness()
{
    if (!stable_) {
        return;
    }
    // pthread_mutex_lock(&last_seen_m_);
    time_t now = time(NULL);
    
    map<string, time_t>::const_iterator it, end;
    for (it = last_seen_.begin(), end = last_seen_.end(); it != end; ++it) {
        TDEBUG("checking %s", it->first.c_str());
        if ((now - it->second) > 10) { // a 10 sec timeout
            // this peer is probably down; i'll try to deprive its membership
            const string &dead = it->first;
            if (dead != primary_) {
                // if it is the primary that is down. there's not much we can do.
                // and if not, we send a request to the primary to update the
                // view
                TINFO("non-primary peer %s is down", dead.c_str());
                DataMessage request;
                request.set_subject("VIEW.REMOVEREQ");
                request.add_dest(primary_);
                request.set_body(dead);
                hermes_->deliver(&request);
            } else {
                TWARN("primary %s is down, now unstable", dead.c_str());
            }

            break;
        }
    }
    // pthread_mutex_unlock(&last_seen_m_);
}

string viewmgr::valueof(unsigned vid, const vector<string> &v)
{
    ostringstream oss;
    oss << vid << " ";
    for (unsigned i = 0; i < v.size(); ++i) {
        oss << v[i] << " ";
    }
    return oss.str();
}

/**
 * Parse a Paxos-decided value into view information.
 * @param pxs_value the value chosen by Paxos
 * @param vid the variable to store parsed vid
 * @param primary the variable to store parsed primary
 * @param cohorts the vector to store parsed cohorts
 */
bool viewmgr::parse_str(const string &pxs_value, unsigned *vid,
        string *primary, vector<string> *cohorts)
{
    istringstream iss(pxs_value);
    if (iss >> *vid) {
        if (iss >> *primary) {
            string s;
            while (iss >> s) {
                cohorts->push_back(s);
            }
            return true;
        }
    }
    return false;
}

/**
 * Switch to a new view and possibly signal the pending joining thread.
 * generally assume view_m_ is held
 */
void viewmgr::switch_view(unsigned vid, const string &primary,
        const vector<string> &cohorts)
{
    if (vid > vid_) {
        TDEBUG("now entering view #%d", vid);
        if (!isamember(me_, cohorts_) && isamember(me_, cohorts)) {
            // i have just joined the network, so notify the possibly pending
            // joining thread 
            pthread_cond_signal(&joined_cv_);
        }
        unsigned oldvid = vid_;
        vid_ = vid;
        primary_ = primary;
        cohorts_ = cohorts;
        last_seen_.clear();
        vector<string>::iterator it = cohorts_.begin(), end = cohorts_.end();
        if (me_ == primary) {
            for (; it != end; ++it) {
                if (*it != me_) last_seen_[*it] = time(NULL);
            }
        } else {
            last_seen_[primary_] = time(NULL);
        }
        stable_ = true;
        pthread_cond_signal(&stable_cv_);
        if (vobserver_) {
            vobserver_->postviewchange(oldvid, vid_, primary, cohorts);
        }
    }
}

void viewmgr::on_message(const DataMessage *data)
{
    string subject = data->subject();
    if (subject == "VIEW.HEARTBEAT") {
        handle_heartbeat(data);
    } else if (subject == "VIEW.JOINREQ") {
        handle_joinreq(data);
    } else if (subject == "VIEW.JOINRES") {
        handle_joinres(data);
    } else if (subject == "VIEW.REMOVEREQ") {
        handle_removereq(data);
    }
}

void viewmgr::handle_heartbeat(const DataMessage *data)
{
    Heartbeat heartbeat;
    if (heartbeat.ParseFromString(data->body())) {
        // update the sender's last seen timestamp
        pthread_mutex_lock(&view_m_);
        const string &src = data->src();
        if (isamember(src, cohorts_)) {
            TDEBUG("heartbeat from %s", data->src().c_str());
            last_seen_[src] = time(NULL);
        } else {
            // this node believes it is one of us, but it is not.
            // reply with my current view
            TWARN("unexpected heartbeat from %s (vid: %d)", src.c_str(),
                    heartbeat.vid());
            DataMessage reply;
            reply.set_subject("VIEW.OUTDATED");
            hermes_->deliver(&reply);
        }
        pthread_mutex_unlock(&view_m_);
    }
}

void viewmgr::handle_joinreq(const DataMessage *data)
{
    pthread_mutex_lock(&view_m_);
    if (primary_ != me_) {
        // i am not the primary. we shouldn't reach here in the current
        // cvmfs setup.
        TWARN("i'm not primary; do not ask me to join");
    } else {
        const string &peer = data->src();
        vc_q_.push_back(pair<string, bool>(peer, true));
        TDEBUG("%s to be joining", peer.c_str());
        pthread_cond_signal(&viewchange_cv_);
    }
    pthread_mutex_unlock(&view_m_);
}

void viewmgr::handle_joinres(const DataMessage *data)
{
    JoinResponse jresp;
    if (jresp.ParseFromString(data->body())) {
        // what if this response comes after the join() has already timed out?
        // there are two possiblities:
        // 1) this node has switched to a newer view, so this response must be
        //    a duplicate and can be safely discarded.
        // 2) this response is a reply to the outstanding join request, but it
        //    comes too late. if we signal the conditional var at this point,
        //    the join() function won't be notified. the recovery thread will
        //    once again, after sleeping for a while, invoke join(), which will 
        //    cause the primary to simply reply with the current membership
        //    information.
        // so the logic of the following code should have no problem but its
        // efficiency can be improved.
        TDEBUG("join response joined %d vid %u myvid: %u", jresp.joined(),
                jresp.vid(), vid_);
        if (jresp.joined()) {
            pthread_mutex_lock(&view_m_);
            if (jresp.vid() > vid_) {
                // a new view has been formed
                TINFO("%s has joined at vid %u", me_.c_str(), jresp.vid());
                vector<string> cohorts;
                cohorts.clear();
                int sz = jresp.cohorts_size();
                for (int i = 0; i < sz; ++i) {
                    cohorts.push_back(jresp.cohorts(i));
                }
                switch_view(jresp.vid(), jresp.primary(), cohorts);
            } else {
                TDEBUG("discarding late join response");
            }
            pthread_mutex_unlock(&view_m_);
        } else {
            TINFO("join req rejected; vid: %u", jresp.vid());
        }
    }
}

void viewmgr::handle_removereq(const DataMessage *data)
{
    pthread_mutex_lock(&view_m_);
    if (primary_ == me_) {
        const string &dead = data->body();
        vc_q_.push_back(pair<string, bool>(dead, false));
        TDEBUG("%s to be removed", dead.c_str());
        pthread_cond_signal(&viewchange_cv_);
    } else {
        TWARN("non-primary cannot handle node removal request");
    }
    pthread_mutex_unlock(&view_m_);
}

